package com.mango.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

@SuppressWarnings({ "deprecation", "unused" })
public class HbaseDao {
	// 声明静态配置
	private Configuration conf = null;
	private Connection conn = null;

	@Before
	public void init() {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "hdf2:2181,hdf3:2181,hdf4:2181");
		// conf.set("hbase.zookeeper.property.clientPort", "2181");//此代码可以要也可以不要
		// 建立一个数据库的连接
		try {
			conn = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Test
	public void mainRunner() throws IOException {
		String[] cols = { "baseInfo", "attachInfo" };
		String table = "TEstTable2";
		// createTable("TEstTable2", cols);
		// insertTable(table);
		String rowKey = "rk0001";
		// getRow(table, rowKey);
		scanTable(table);
	}

	public HBaseAdmin getAdmin() {
		HBaseAdmin hAdmin = null;
		try {
			// 创建一个数据库管理员
			hAdmin = (HBaseAdmin) conn.getAdmin();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return hAdmin;
	}

	@After
	public void closeConn() {
		if (conn != null && !conn.isClosed())
			try {
				conn.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				System.out.println("conn faile to close");
			}
	}

	// 创建数据库表
	public void createTable(String tableName, String[] columnFamilys) throws IOException {
		// 创建一个数据库管理员
		HBaseAdmin hAdmin = getAdmin();
		TableName tn = TableName.valueOf(tableName);
		if (hAdmin.tableExists(tn)) {
			System.out.println(tableName + "表已存在");
			System.exit(0);
		} else {
			// 新建一个表描述
			HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
			// 在表描述里添加列族
			for (String columnFamily : columnFamilys) {
				HColumnDescriptor col = new HColumnDescriptor(columnFamily);
				col.setMaxVersions(5);
				tableDesc.addFamily(col);
			}
			// 根据配置好的表描述建表
			hAdmin.createTable(tableDesc);
			System.out.println("创建" + tableName + "表成功");
		}
		// 关闭连接
		closeConn();
	}

	public void insertTable(String tablename) throws IOException {
		HBaseAdmin admin = getAdmin();
		Put name = new Put(Bytes.toBytes("rk0001"));
		HTable table = (HTable) conn.getTable(TableName.valueOf(tablename));
		name.addColumn(Bytes.toBytes("baseInfo"), Bytes.toBytes("name"), Bytes.toBytes("angelabay"));
		try {
			table.put(name);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.out.println("插入数据失败");
		}
		System.out.println("插入成功");
	}

	// 通过rowkey获取一条数据（获取指定行数据）
	public void getRow(String tableName, String rowKey) throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(rowKey));
		// 输出结果
		Result result = table.get(get);
		for (Cell cell : result.rawCells()) {
			System.out.println("行键:" + new String(CellUtil.cloneRow(cell)) + "\t" + "列族:"
					+ new String(CellUtil.cloneFamily(cell)) + "\t" + "列名:" + new String(CellUtil.cloneQualifier(cell))
					+ "\t" + "值:" + new String(CellUtil.cloneValue(cell)) + "\t" + "时间戳:" + cell.getTimestamp());
		}
		// 关闭资源
		table.close();
	}

	// 扫描全表
	public void scanTable(String tableName) {
		// 获取表
		try {
			HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
			// 创建一个扫描对象
			Scan scan = new Scan();
			// 扫描全表输出结果
			ResultScanner results = table.getScanner(scan);
			for (Result result : results) {
				for (Cell cell : result.rawCells()) {
					System.out.println("行键:" + new String(CellUtil.cloneRow(cell)) + "\t" + "列族:"
							+ new String(CellUtil.cloneFamily(cell)) + "\t" + "列名:"
							+ new String(CellUtil.cloneQualifier(cell)) + "\t" + "值:"
							+ new String(CellUtil.cloneValue(cell)) + "\t" + "时间戳:" + cell.getTimestamp());
				}
			}
			// 关闭资源
			results.close();
			table.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	// 删除指定行键的一条数据
	public void delRow(String tableName, String rowKey) throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 删除数据
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		table.delete(delete);
		// 关闭资源
		table.close();
	}

	// 删除指定行键的多条数据
	public void delRow(String tableName, String[] rows) throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 删除多条数据
		List<Delete> list = new ArrayList<Delete>();
		for (String row : rows) {
			Delete delete = new Delete(Bytes.toBytes(row));
			list.add(delete);
		}
		table.delete(list);
		// 关闭资源
		table.close();
	}

	// 删除列族
	public void delColumnFamily(String tableName, String columnFamily) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
		admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(columnFamily));
	}

	// 删除数据库表
	public void deleteTable(String tableName) throws IOException {
		// 创建一个数据库管理员
		HBaseAdmin hAdmin = getAdmin();
		if (hAdmin.tableExists(TableName.valueOf(tableName))) {
			// 失效表
			hAdmin.disableTable(TableName.valueOf(tableName));
			// 删除表
			hAdmin.deleteTable(TableName.valueOf(tableName));
			System.out.println("删除" + tableName + "表成功");
			conn.close();
		} else {
			System.out.println("需要删除的" + tableName + "表不存在");
			System.exit(0);
		}
	}

	// 追加插入(将原有value的后面追加新的value，如原有value=a追加value=bc则最后的value=abc)
	public void appendData(String tableName, String rowKey, String columnFamily, String column, String value)
			throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 通过rowkey创建一个append对象
		Append append = new Append(Bytes.toBytes(rowKey));
		// 在append对象中设置列族、列、值
		append.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
		// 追加数据
		table.append(append);
		// 关闭资源
		table.close();
	}

	// 计数器(amount为正数则计数器加，为负数则计数器减，为0则获取当前计数器的值)
	public long incrementColumnValue(String tableName, String rowKey, String columnFamily, String column, long amount)
			throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 计数器
		long result = table.incrementColumnValue(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamily),
				Bytes.toBytes(column), amount);
		// 关闭资源
		table.close();
		return result;
	}

	// 符合条件后添加数据(只能针对某一个rowkey进行原子操作)
	public boolean checkAndPut(String tableName, String rowKey, String columnFamilyCheck, String columnCheck,
			String valueCheck, String columnFamily, String column, String value) throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 设置需要添加的数据
		Put put = new Put(Bytes.toBytes(rowKey));
		put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
		// 当判断条件为真时添加数据
		boolean result = table.checkAndPut(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamilyCheck),
				Bytes.toBytes(columnCheck), Bytes.toBytes(valueCheck), put);
		// 关闭资源
		table.close();
		return result;
	}

	// 符合条件后刪除数据(只能针对某一个rowkey进行原子操作)
	public boolean checkAndDelete(String tableName, String rowKey, String columnFamilyCheck, String columnCheck,
			String valueCheck, String columnFamily, String column) throws IOException {
		// 获取表
		HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
		// 设置需要刪除的delete对象
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		delete.addColumn(Bytes.toBytes(columnFamilyCheck), Bytes.toBytes(columnCheck));
		// 当判断条件为真时添加数据
		boolean result = table.checkAndDelete(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamilyCheck),
				Bytes.toBytes(columnCheck), Bytes.toBytes(valueCheck), delete);
		// 关闭资源
		table.close();
		return result;
	}
}
